[Kernel-Spark] Fix startingVersion support on non re-creatable version by skipping start snapshot validation #5574
+152
−23
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ad time This change fixes the startingVersion support in DSv2 streaming by: 1. Removing the requirement for initialSnapshot parameter 2. Using getActionsFromCommitFilesWithProtocolValidation to validate protocol for each commit at read time (similar to V1 behavior) 3. Extracting getActionsUnsafe utility method to StreamingHelper The key insight is that we don't need a pre-loaded snapshot for protocol validation - we can validate each commit's protocol as we read it, which matches V1's behavior and avoids the issue where initialSnapshot.version < startVersion. Test: DeltaSourceDSv2Suite startingVersion tests now pass
huan233usc
changed the title
huan233usc
changed the title
huan233usc
changed the title
zikangh
reviewed
Nov 27, 2025
| return removeFile.getDataChange() ? Optional.of(removeFile) : Optional.empty(); | ||
| } | ||
|
|
||
| public static CloseableIterator<ColumnarBatch> getActionsFromRangeUnsafe( |
Contributor
There was a problem hiding this comment.
Could you add a comment explain in what way this is unsafe?
|
|
||
| public SparkMicroBatchStream( | ||
| DeltaSnapshotManager snapshotManager, | ||
| Snapshot initialSnapshot, |
Contributor
There was a problem hiding this comment.
we don't necessarily need to plumb in an initialSnapshot, right?
Collaborator
Author
There was a problem hiding this comment.
I need to get the table path and pass to ValidateCommit method. Let me just plumb the path instead
Collaborator
Author
There was a problem hiding this comment.
Actually after rebasing I found that we are trying to load the latest snapshot again. So we may want to plumb it so that we could get snapshot, table id and path without getting snapshot again
- Resolved conflicts in SparkMicroBatchStream.java and SparkMicroBatchStreamTest.java - Updated constructor calls to use tablePath parameter instead of initialSnapshot - Applied master's improvements (try-catch for CommitRangeNotFoundException, Java Optional usage)
huan233usc
requested review from
gengliangwang and
zikangh
and removed request for
zikangh
zikangh
approved these changes
Dec 3, 2025
|
|
||
| public SparkScan( | ||
| DeltaSnapshotManager snapshotManager, | ||
| io.delta.kernel.Snapshot initialSnapshot, |
Contributor
There was a problem hiding this comment.
do we need to fully quality here?
Collaborator
There was a problem hiding this comment.
+1, let's avoid this
| String tablePath = startSnapshot.getPath(); | ||
| try (CloseableIterator<ColumnarBatch> actionsIter = | ||
| commitRange.getActions(engine, startSnapshot, ACTION_SET)) { | ||
| StreamingHelper.getActionsFromRangeUnsafe( |
Contributor
There was a problem hiding this comment.
Could you add a comment here documenting the exact use case for why we are calling getActionsFromRangeUnsafe?
gengliangwang
approved these changes
Dec 3, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Which Delta project/connector is this regarding?
Description
This PR fixes the
startingVersionsupport in DSv2 streaming by skipping start snapshot validation when callinggetActions.Problem
The Kernel API's
CommitRange.getActions()strictly requiredsnapshot.version == startVersion, which failed when:startingVersion="latest"and no new data exists (effective version =latestVersion + 1)Solution
Use low-level Kernel API: Call
DeltaLogActionUtils.getActionsFromCommitFilesWithProtocolValidation()directly instead ofCommitRange.getActions(), bypassing the strict snapshot version checkWhether we want to update kernel API to allow us bypass validation worth longer discussion. We just made this change in Dsv2 connector to make sure v1 behavior parity.
How was this patch tested?
Unit
Does this PR introduce any user-facing changes?
No